package com.duowan.realtime.computing;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.thrift.TException;
import org.springframework.util.Assert;

import com.duowan.realtime.computing.util.BeanFactory;
import com.duowan.realtime.thirft.api.BloomFilterException;
import com.duowan.realtime.thirft.api.BloomFilterGroupQuery;
import com.duowan.realtime.thirft.api.DistinctData;
import com.duowan.realtime.thirft.api.DistinctRequest;
import com.duowan.realtime.webservice.DataComputingWebService;

public class BloomFilterClient {

	private final static Map<String,RawBloomFilterClient> bloomfilterClientMap = new ConcurrentHashMap<String, RawBloomFilterClient>();
	
	private DataComputingWebService dataComputingWebService;
	

	/**
	 * 默认构造函数
	 */
	public BloomFilterClient(){
		setDataComputingWebService(BeanFactory.getDataComputingWebService());
	}
	
	public void setDataComputingWebService(DataComputingWebService dataComputingWebService) {
		this.dataComputingWebService = dataComputingWebService;
	}

	public Map<String, Integer> notContainsCountAndAdd(String bloomfilterGroup, List<BloomFilterGroupQuery> bfGroupQuery) throws BloomFilterException, TException {
		RawBloomFilterClient client = getClientByBloomfilterGroup(bloomfilterGroup);
		return client.notContainsCountAndAdd(bloomfilterGroup, bfGroupQuery);
	}
	
	public Map<String, List<DistinctData>> notContainsAndMark(String bloomfilterGroup, List<DistinctRequest> bfCounterRequest) throws BloomFilterException, TException {
		RawBloomFilterClient client = getClientByBloomfilterGroup(bloomfilterGroup);
		return client.notContainsAndMark(bloomfilterGroup, bfCounterRequest);
	}
	
	public RawBloomFilterClient getClientByBloomfilterGroup(String bloomfilterGroup) {
		RawBloomFilterClient client = bloomfilterClientMap.get(bloomfilterGroup);
		if(client == null) {
			String serverIp = dataComputingWebService.getServerIp(bloomfilterGroup);
			Assert.hasText(serverIp,"not found serverIp for bloomfilterGroup:"+bloomfilterGroup);
			client = new RawBloomFilterClient();
			client.setHost(serverIp);
			try {
				client.afterPropertiesSet();
			} catch (Exception e) {
				new RuntimeException("afterPropertiesSet error:", e);
			}
			bloomfilterClientMap.put(bloomfilterGroup, client);
		}
		return client;
	}

}
